 1.Spark原理之作业执行原理中Shuffle详解下Spark Shuffle的两个阶段
   
   在 Spark 或 MapReduce 分布式计算框架中，数据被分成一块一块的分区，分布在集群中各节点上，每个计算任务
一次处理一个分区，当需要对具有某种共同特征的一类数据进行计算时，就需要将集群中的这类数据汇聚到同一节
点。这个按照一定的规则对数据重新分区的过程就是Shuffle。
   对于Spark来讲，一些Transformation或Action算子会让RDD产生宽依赖，即Parent RDD中的每个Partition被child
RDD中的多个Partition使用，这时需要进行Shuffle，根据Record的key对parent RDD进行重新分区。
   以Shuffle为边界,Spark将一个Job划分为不同的Stage。Spark的Shuffle分为Write和Read两个阶段，分属于两个不
同的Stage，前者是Parent Stage的最后一步，后者是Child Stage的第一步。
   Spark 的 Stage 分为两种：
       ResultStage。负责返回计算结果
       ShuffleMapStage。其他的均为ShuffleMapStage
   如果按照 map 端和 reduce 端来分析的话：
       ShuffleMapStage可以即是map端任务，又是reduce端任务
       ResultStage只能充当reduce端任务 
   Spark Shuffle的流程简单抽象为以下几步：
       Shuffle Write
	       Map side combine (if needed)
           Write to local output file
	   Shuffle Read
	       Block fetch
           Reduce side combine
           Sort (if needed)
 
 2.Spark Shuffle 技术演进
   
   在Spark Shuffle的实现上，经历了Hash、Sort、Tungsten-Sort三阶段：
       Spark 0.8及以前 Hash Based Shuffle
       Spark 0.8.1 为Hash Based Shuffle引入File Consolidation机制
       Spark 0.9 引入ExternalAppendOnlyMap
       Spark 1.1 引入Sort Based Shuffle，但默认仍为Hash Based Shuffle
       Spark 1.2 默认的Shuffle方式改为Sort Based Shuffle
       Spark 1.4 引入Tungsten-Sort Based Shuffle
       Spark 1.6 Tungsten-sort并入Sort Based Shuffle
       Spark 2.0 Hash Based Shuffle退出历史舞台
   简单的说：
       Spark 1.1 以前是Hash Shuffle
       Spark 1.1 引入了Sort Shuffle
       Spark 1.6 将Tungsten-sort并入Sort Shuffle（利用对外内存进行排序）
       Spark 2.0 Hash Shuffle退出历史舞台
 
 3.Hash Shuffle V1
   
   相对于传统的 MapReduce，Spark 假定大多数情况下 Shuffle 的数据不需要排序，强制排序反而会降低性能。因此
不在 Shuffle Read 时做 Merge Sort，如果需要合并的操作的话，则会使用聚合(agggregator)，即用了一个HashMap 
(实际上是一个 AppendOnlyMap)来将数据进行合并
   在 Map Task 过程按照 Hash 的方式重组 Partition 的数据，不进行排序。每个 Map Task 为每个 Reduce Task 生成
一个文件，通常会产生大量的文件（即对应为 M*R 个中间文件，其中 M 表示 Map Task 个数，R 表示 Reduce Task个数)
，伴随大量的随机磁盘 I/O 操作与大量的内存开销。
   Hash Shuffle V1的两个严重问题：
       生成大量文件，占用文件描述符，同时引入DiskObjectWriter带来的Writer Handler 的缓存也非常消耗内存
	   如果在 Reduce Task 时需要合并操作的话，会把数据放在一个 HashMap 中进行合并，如果数据量较大，很容
易引发 OOM

 4.Hash Shuffle V2 -- File Consolidation
   
   针对上面的第一个问题，Spark 做了改进，引入了 File Consolidation 机制。
   一个 Executor 上所有的 Map Task 生成的分区文件只有一份，即将所有的 Map Task 相同的分区文件合并，这样每
个 Executor 上最多只生成 N 个分区文件。
   这样减少了文件数，但是假如下游 Stage 的分区数 N 很大，还是会在每个 Executor 上生成 N 个文件，同样，如果
一个 Executor 上有 K 个 Core，还是会开 K*N 个 Writer Handler，这里仍然容易导致OOM。

 5.Sort Shuffle V1
   
   为了更好地解决上面的问题，Spark 参考了 MapReduce 中 Shuffle 的处理方式，引入基于排序的 Shuffle 写操作机
制。
   每个 Task 不会为后续的每个 Task 创建单独的文件，而是将所有对结果写入同一个文件。该文件中的记录首先是按照
Partition Id 排序，每个 Partition 内部再按照 Key 进行排序，Map Task 运行期间会顺序写每个 Partition 的数据，
同时生成一个索引文件记录每个 Partition 的大小和偏移量。
   在 Reduce 阶段，Reduce Task 拉取数据做 Combine 时不再采用 HashMap，而是采用ExternalAppendOnlyMap，
该数据结构在做 Combine 时，如果内存不足，会刷写磁盘，避免大数据情况下的 OOM。
   总体上看来 Sort Shuffle 解决了 Hash Shuffle 的所有弊端，但是因为需要其 Shuffle 过程需要对记录进行排序，所
以在性能上有所损失。
   Tungsten-Sort Based Shuffle / Unsafe Shuffle
   从 Spark 1.5.0 开始，Spark 开始了钨丝计划（Tungsten），目的是优化内存和CPU的使用，进一步提升Spark的性能。
由于使用了堆外内存，而它基于 JDK Sun Unsafe API，故 Tungsten-Sort Based Shuffle 也被称为 Unsafe Shuffle。
   它的做法是将数据记录用二进制的方式存储，直接在序列化的二进制数据上 Sort 而不是在 Java 对象上，这样一方面
可以减少内存的使用和 GC 的开销，另一方面避免 Shuffle 过程中频繁的序列化以及反序列化。在排序过程中，它提供 
cache-efficient sorter，使用一个 8 bytes 的指针，把排序转化成了一个指针数组的排序，极大的优化了排序性能。
   但是使用 Tungsten-Sort Based Shuffle 有几个限制，Shuffle 阶段不能有 aggregate 操作，分区数不能超过一定
大小(2^24-1,这是可编码的最大 Parition Id)，所以像 reduceByKey 这类有 aggregate 操作的算子是不能使用
Tungsten-Sort Based Shuffle，它会退化采用 Sort Shuffle。

 6.Sort Shuffle V2
   
   从 Spark1.6.0 开始，把 Sort Shuffle 和 Tungsten-Sort Based Shuffle 全部统一到 Sort Shuffle 中，如果检测到满足
Tungsten-Sort Based Shuffle 条件会自动采用 Tungsten-Sort Based Shuffle，否则采用 Sort Shuffle。
   从Spark2.0 开始，Spark 把 Hash Shuffle 移除， Spark2.x 中只有一种 Shuffle，即为 Sort Shuffle。
   